/*
 * 文件名： KafkaValueEncoder.java
 * 
 * 创建日期： 2017年3月14日
 *
 * Copyright(C) 2017, by <a href="mailto:liws@xingyuanauto.com">liws</a>.
 *
 * 原始作者: liws
 *
 */
package com.moli.iov.protocol.codec;

import java.util.Date;
import java.util.Map;

import com.moli.iov.protocol.codec.ReqStreamData;
import com.moli.iov.protocol.enums.DateTimeTag;

import kafka.utils.VerifiableProperties;
import org.apache.kafka.common.serialization.Serializer;

/**
 * 消息上报 kafka消息value编码器
 *
 * @author <a href="mailto:liws@xingyuanauto.com">liws</a>
 *
 * @version $Revision$
 *
 * @since 2017年3月14日
 */
public class KafkaValueEncoder implements Serializer<ReqStreamData> {

	public KafkaValueEncoder() {

	}

	public KafkaValueEncoder(VerifiableProperties verifiableProperties) {

	}

	/**
	 *
	 * @see kafka.serializer.Encoder#toBytes(Object)
	 */
//	@Override
//	public byte[] toBytes(ReqStreamData reqStreamData) {
//        byte[] dataBytes = reqStreamData.getDataBytes();//原始报文
//        Date reportTime = reqStreamData.getReqDateTime();//接收时间
//        DateTimeTag tag = new DateTimeTag(reportTime);
//        byte[] tBytes = tag.toBytes();
//        byte[] rBytes = new byte[dataBytes.length + tBytes.length];
//        System.arraycopy(tBytes, 0, rBytes, 0, tBytes.length);
//        System.arraycopy(dataBytes, 0, rBytes, tBytes.length, dataBytes.length);
//        return rBytes;
////		return SerializeUtils.serialize(reqData);
//	}

	public static void main(String[] args) {
	    ReqStreamData reqData = new ReqStreamData(new Date(), new byte[]{0x01, 0x02,0x03});
	    KafkaValueEncoder kafkaValueEncoder = new KafkaValueEncoder();
//	    byte[] tByte = kafkaValueEncoder.toBytes(reqData);

	}

	@Override
	public void configure(Map<String, ?> configs, boolean isKey) {

	}

	@Override
	public byte[] serialize(String topic, ReqStreamData data) {
		byte[] dataBytes = data.getDataBytes();//原始报文
		Date collectDate = data.getCollectDate();//接收时间
		DateTimeTag tag = new DateTimeTag(collectDate);
		byte[] tBytes = tag.toBytes();
		byte[] rBytes = new byte[dataBytes.length + tBytes.length];
		System.arraycopy(tBytes, 0, rBytes, 0, tBytes.length);
		System.arraycopy(dataBytes, 0, rBytes, tBytes.length, dataBytes.length);
		return rBytes;
	}

	@Override
	public void close() {

	}
}